package com.pass.one;

import java.io.IOException;
import java.util.HashSet;
import java.util.Set;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import com.data.ConnectedEdge;
import com.data.Node;
import com.join.SpatialJoin;
import com.uf.UnionFind;

/**
 * Pass One Reducer gets the output of Pass One Mapper, puts the Nodes in the
 * same group together, does spatial join for all the nodes in one group and
 * performs union find on the edges. The output format is <GroupID, edge edge
 * edge....>, where edges are edges that is done with union find that only
 * involve intersection nodes.
 * 
 * @author bh349
 */
public class PassOneReducer extends
		Reducer<IntWritable, Text, IntWritable, Text> {
	@Override
	public void reduce(IntWritable key, Iterable<Text> values, Context context)
			throws IOException, InterruptedException {
		Set<Node> nodes = new HashSet<Node>();
		for (Text value : values) {
			nodes.add(new Node(value.toString()));
		}
		Set<ConnectedEdge> edgesBeforeUF = SpatialJoin.spatialJoin(nodes);

		// get union find result
		Set<ConnectedEdge> edges = (new UnionFind(edgesBeforeUF)).getUFEdges();
		// get only edges that involve intersection nodes.
		Set<ConnectedEdge> removedEdges = new HashSet<ConnectedEdge>();
		for (ConnectedEdge edge : edges) {
			if (!edge.getNode1().isIntersection()
					&& !edge.getNode2().isIntersection()) {
				removedEdges.add(edge);
			}
		}
		edges.removeAll(removedEdges);
		for (ConnectedEdge edge : edges) {
			context.write(key, new Text(edge.toString()));
		}
	}
}
